package com.central.controller
import com.central.service.GangFindService
import com.central.service.GangFindCondition
import com.central.utils.ConfigUtil
import com.typesafe.config.Config
import org.apache.spark.SparkContext
import org.apache.spark.sql.SparkSession
import org.scalactic.Accumulation.withGood
import org.scalactic._
import spark.jobserver.api.{JobEnvironment, SingleProblem, SparkJob, ValidationProblem}

import scala.util.{Failure, Success, Try}

object GangController extends SparkJob{
  override type JobData = (SparkSession,GangFindCondition)
  override type JobOutput = Unit

  override def runJob(sc: SparkContext, runtime: JobEnvironment, data: JobData): JobOutput = {
    new GangFindService().findCrime(data._1,data._2)
  }

  override def validate(sc: SparkContext, runtime: JobEnvironment, config: Config):  JobData Or Every[ValidationProblem] = {
    val spark = SparkSession
      .builder()
      .appName("gangAnalysis")
      .master("spark://192.168.5.180:7077")
      .config("es.nodes", ConfigUtil.properties.getProperty("elasticsearch.host", "192.168.5.180"))
      .config("es.port", ConfigUtil.properties.getProperty("elasticsearch.port", "9200"))
      .config("es.mapping.date.rich", "false")
      .getOrCreate()

    var conditions: GangFindCondition = null
    val parameterValidatin: (SparkSession, GangFindCondition) Or One[ValidationProblem] =
      Try { //将传入的参数转换成GangFindCondition
        config.getConfig("body")
      } match {
        case Success(config) => {
          val jobid = config.getString("jobid")
          val srctype = config.getString("srctype")
          val srccode = config.getString("srccode")
          val activitytime = config.getString("activitytime")
          val endTime = config.getString("endtime")
          val starttime = config.getString("starttime")
          val devicecodes = config.getString("devicecodes")
          val hascriminalrecord = config.getInt("hascriminalrecord")
          val hashisrelation = config.getInt("hashisrelation")
          val accumulatedays = config.getInt("accumulatedays")
          val hasliferow = config.getInt("hasliferow")


           conditions = GangFindCondition(srccode,srctype,jobid,devicecodes,activitytime,starttime,endTime,hashisrelation,hascriminalrecord,hasliferow,accumulatedays)
          Good((spark, conditions))
        }
        case Failure(ex) => Bad(One(SingleProblem("body is not exists or json can't parse. error message:" + ex.getMessage)))
      }
    withGood(parameterValidatin) { a => a }

  }
}
